package com.gitee.ixtf.rsocket.verticle

import com.gitee.ixtf.rsocket.kotlinx.cloudEvent
import com.gitee.ixtf.rsocket.service.PONG
import com.gitee.ixtf.rsocket.service.PingRequest
import io.cloudevents.CloudEvent
import io.rsocket.Payload
import io.rsocket.RSocket
import io.rsocket.util.DefaultPayload
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.reactive.asFlow
import kotlinx.coroutines.reactor.asFlux
import kotlinx.coroutines.reactor.mono
import org.reactivestreams.Publisher
import reactor.core.publisher.Flux
import reactor.core.publisher.Mono

internal interface CloudEventHandler : RSocket {

  suspend fun fireAndForget(event: CloudEvent): Unit = TODO()

  suspend fun requestResponse(event: CloudEvent): Payload = TODO()

  suspend fun requestStream(event: CloudEvent): Flow<Payload> = TODO()

  suspend fun requestChannel(events: Flow<CloudEvent>): Flow<Payload> = TODO()

  suspend fun metadataPush(event: CloudEvent): Unit = TODO()

  override fun fireAndForget(payload: Payload): Mono<Void> =
      mono { fireAndForget(payload.cloudEvent()) }.then()

  override fun requestResponse(payload: Payload): Mono<Payload> = mono {
    val event = payload.cloudEvent()
    when (event.type) {
      PingRequest.CLOUD_EVENT_TYPE -> DefaultPayload.create(PONG)
      else -> requestResponse(event)
    }
  }

  override fun requestStream(payload: Payload): Flux<Payload> =
      mono { requestStream(payload.cloudEvent()) }.flatMapMany { it.asFlux() }

  override fun requestChannel(payloads: Publisher<Payload>): Flux<Payload> =
      mono { requestChannel(payloads.asFlow().map { it.cloudEvent() }) }.flatMapMany { it.asFlux() }

  override fun metadataPush(payload: Payload): Mono<Void> =
      mono { metadataPush(payload.cloudEvent()) }.then()
}
